package com.zhuangjie.rabbit.send.wrap;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

@Component
@Slf4j
public class MQHelper {

    @Autowired
    private ApplicationContext applicationContext;



    /**
     *  发送MQ消息并且进行超时等待
     *
     * @param exchangeName 交换机名称
     * @param routingKey   路由密钥
     * @param message      消息
     * @param timeout
     * @param timeUnit
     * @throws Exception 例外
     */
    public void send(String exchangeName, String routingKey, Object message,Long timeout, TimeUnit timeUnit) throws Exception {
        if ((timeout == null && timeUnit != null) || (timeout != null && timeUnit == null)) {
            throw new Exception("超时时间或超时单位其中一个为空，这可能不是你想要的调用！");
        }
        AtomicReference<Boolean> isSendSuccessRef = new AtomicReference<>(false);
        AtomicReference<String> failCause = new AtomicReference<>("");
        Object lock = new Object();
        synchronized (lock) {
            RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
            rabbitTemplate.setConfirmCallback((CorrelationData correlationData, boolean ack, String cause) -> {
                if (ack) {
                    isSendSuccessRef.set(true);
                }else {
                    failCause.set(cause);
                }
                synchronized (lock) {
                    lock.notifyAll();
                }
            });
            rabbitTemplate.convertAndSend(exchangeName, routingKey, message);
            if (timeout != null && timeUnit != null ) {
                lock.wait(timeUnit.toMillis(timeout));
            }else {
                lock.wait();
            }
        }
        if (!isSendSuccessRef.get()) {
            log.warn("mq: 消息发送失败，原因{}.", failCause.get());
            throw new Exception("MQ发送失败,原因："+failCause.get());
        }
    }

    /**
     * 发送MQ消息并且不超时阻塞等待处理结果
     *
     * @param exchangeName 交换机名称
     * @param routingKey   路由密钥
     * @param message      消息
     * @throws Exception 例外
     */
    public void send(String exchangeName, String routingKey, Object message) throws Exception {
        send(exchangeName, routingKey, message, null, null);
    }

}
